
#include "core/threadpool.h"
#include "core/timer.h"

// 1ms ? FIXME: this definition is not rigorous
static utils::microseconds execution_time_slice(1000);

void ThreadPool::ThreadWork(void) {
  try {
    while(true) {
      boost::mutex::scoped_lock lock(mutex_);
      while (job_queue_.empty())
        wait_job_cond_.wait(lock);

      Execution execution = job_queue_.front();
      job_queue_.pop_front();
      lock.unlock();

      DLOG(INFO) << boost::this_thread::get_id() << " get a job";
      utils::microseconds time_slice(execution_time_slice);
      while (true) {
        auto now = utils::steady_clock::now();
        bool finish = execution();
        utils::microseconds time_consume = 
            utils::duration_cast<utils::microseconds>(
            utils::steady_clock::now() - now);

        DLOG(INFO) << boost::this_thread::get_id() 
                  << ": time consume " << time_consume.count() 
                  << " microseconds";
        if (finish) {
          break;
        }
        else if(time_slice <= time_consume) {
          Add(execution);
          DLOG(INFO) << boost::this_thread::get_id() 
                    << ": context switch";
          break;
        }

        DLOG(INFO) << boost::this_thread::get_id() << ": continue";
        time_slice -= time_consume;
      }
      boost::this_thread::interruption_point();
    }
  } catch (boost::thread_interrupted&) {}
}

ThreadPool::ThreadPool(void) : job_queue_(5), concurrency_(0) {}

void ThreadPool::Run(uint32_t concurrency_hint) {
  uint32_t c = boost::thread::hardware_concurrency();
  c = c == 0 ? 4 : (c * 2 + 2);
  if (concurrency_hint != 0 && concurrency_hint < c)
    c = concurrency_hint;

  concurrency_ = c;
  threads_.reset(new boost::thread [c]);
  for (uint32_t i = 0; i < c; i++)
    threads_[i] = boost::thread(boost::bind(&ThreadPool::ThreadWork, this));

  uint32_t feq = boost::lexical_cast<uint32_t>(
      Config::Get("server.timer_feq"));
  utils::milliseconds interval(1000 / feq);
  boost::thread::id not_a_thread;

  while(true) {
    auto start = utils::steady_clock::now();
    Timers::Instance().Tick();

    uint32_t finalize_threads = 0;
    for (uint32_t i = 0; i < c; i++) {
      if (threads_[i].get_id() == not_a_thread ||
          threads_[i].try_join_until(utils::steady_clock::now())) {
        finalize_threads++;
      }
    }

    if (finalize_threads == c)
      break;

    utils::milliseconds time_cost = utils::duration_cast<utils::milliseconds>(
        utils::steady_clock::now() - start);
    if (interval > time_cost)
      boost::this_thread::sleep_for(interval - time_cost);
  }
}

void ThreadPool::Add(const Execution& execution) {
  boost::mutex::scoped_lock l(mutex_);
  if (job_queue_.full()) 
    job_queue_.set_capacity(job_queue_.capacity() * 2);

  job_queue_.push_back(execution);
  l.unlock();
  wait_job_cond_.notify_one();
}

void ThreadPool::Finalize(void) {
  for (uint32_t i = 0; i < concurrency_; i++)
    threads_[i].interrupt();
}

